from __future__ import annotations

import json
from pathlib import Path
from typing import Dict, List

import matplotlib
matplotlib.use("Agg")  # 使用非GUI后端，支持无头环境/Windows环境
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_absolute_error, r2_score
from sklearn.model_selection import train_test_split

# 定义项目根目录和数据路径
PROJECT_ROOT = Path(__file__).parent
DATA_PATH = PROJECT_ROOT / "data" / "second_hand_housing.csv"
OUTPUT_DIR = PROJECT_ROOT / "outputs"


def load_data(csv_path: Path) -> pd.DataFrame:
    """
    加载原始CSV数据集并确保数值列正确解析
    
    Args:
        csv_path: CSV文件的路径
        
    Returns:
        加载后的DataFrame
    """
    # 读取CSV文件
    df = pd.read_csv(csv_path)
    
    # 获取除'location'外的所有数值列
    numeric_cols = df.columns.difference(["location"])
    
    # 将数值列转换为数值类型，无法转换的设为NaN
    for column in numeric_cols:
        df[column] = pd.to_numeric(df[column], errors="coerce")
    
    return df


def clean_data(df: pd.DataFrame) -> tuple[pd.DataFrame, Dict[str, int]]:
    """
    清理数据集：删除重复值并填充缺失值
    
    Args:
        df: 原始数据DataFrame
        
    Returns:
        tuple: 包含清理后的DataFrame和缺失值统计字典
    """
    # 创建数据副本，避免修改原始数据
    cleaned = df.copy()
    
    # 步骤1: 删除完全重复的行
    cleaned = cleaned.drop_duplicates()
    print(f"删除了 {len(df) - len(cleaned)} 个重复行")

    # 记录清理前的缺失值情况
    missing_before = cleaned.isna().sum().to_dict()

    # 步骤2: 处理数值列的缺失值
    numeric_cols = cleaned.columns.difference(["location"])
    for column in numeric_cols:
        # 计算每列的中位数
        median_value = cleaned[column].median()
        # 用中位数填充缺失值
        cleaned[column] = cleaned[column].fillna(median_value)
        print(f"列 '{column}': 用中位数 {median_value:.2f} 填充了 {missing_before[column]} 个缺失值")

    # 步骤3: 处理分类列'location'的缺失值
    if "location" in cleaned.columns:
        # 计算众数（最常出现的位置）
        mode_location = cleaned["location"].mode()
        if not mode_location.empty:
            # 用众数填充缺失值
            location_missing_before = missing_before.get("location", 0)
            cleaned["location"] = cleaned["location"].fillna(mode_location.iloc[0])
            print(f"列 'location': 用众数 '{mode_location.iloc[0]}' 填充了 {location_missing_before} 个缺失值")

    # 将location列转换为分类类型以节省内存
    cleaned["location"] = cleaned["location"].astype("category")
    
    # 步骤4: 创建新特征 - 楼层比例
    cleaned["floor_ratio"] = cleaned["floor"] / cleaned["total_floors"]
    print("创建了新特征 'floor_ratio' (floor / total_floors)")

    # 记录清理后的缺失值情况
    missing_after = cleaned.isna().sum().to_dict()

    return cleaned, {"before": missing_before, "after": missing_after}


def compute_basic_statistics(df: pd.DataFrame) -> pd.DataFrame:
        """
        因客户要求，该部分业务逻辑已折叠
        """


def generate_visualisations(df: pd.DataFrame) -> None:
    """
    生成并保存探索性数据分析(EDA)图表
    
    Args:
        df: 清理后的DataFrame
    """
    # 确保输出目录存在
    OUTPUT_DIR.mkdir(exist_ok=True, parents=True)
    print("开始生成可视化图表...")

    # 只选择数值列进行相关性分析
    corr_df = df.select_dtypes(include=[np.number])
    
    # 检查是否包含目标变量'price'
    if "price" not in corr_df.columns:
        raise ValueError("数据框中缺少'price'列")

    # 图表1: 特征相关性热力图
    plt.figure(figsize=(10, 8))
    sns.heatmap(corr_df.corr(), annot=True, fmt=".2f", cmap="RdYlBu", square=True)
    plt.title("特征相关性热力图")
    plt.tight_layout()
    plt.savefig(OUTPUT_DIR / "correlation_heatmap.png", dpi=150)
    plt.close()
    print("已保存: correlation_heatmap.png")

    # 图表2: 面积vs价格散点图（按位置着色）
    plt.figure(figsize=(8, 6))
    sns.scatterplot(data=df, x="area", y="price", hue="location", palette="Set2", s=70)
    plt.title("面积 vs. 价格（按位置）")
    plt.xlabel("面积 (平方米)")
    plt.ylabel("价格 (人民币)")
    plt.tight_layout()
    plt.savefig(OUTPUT_DIR / "area_vs_price.png", dpi=150)
    plt.close()
    print("已保存: area_vs_price.png")

    # 图表3: 地铁距离vs价格散点图（按位置着色）
    plt.figure(figsize=(8, 6))
    sns.scatterplot(data=df, x="distance_to_metro", y="price", hue="location", palette="Set1", s=70)
    plt.title("地铁距离 vs. 价格")
    plt.xlabel("距离地铁距离 (公里)")
    plt.ylabel("价格 (人民币)")
    plt.tight_layout()
    plt.savefig(OUTPUT_DIR / "metro_distance_vs_price.png", dpi=150)
    plt.close()
    print("已保存: metro_distance_vs_price.png")


def train_and_evaluate(df: pd.DataFrame) -> Dict[str, float]:
    """
    使用面积特征训练简单线性回归模型预测价格，并评估模型性能
    
    Args:
        df: 清理后的DataFrame
        
    Returns:
        包含模型评估指标的字典
    """
    # 准备特征和目标变量
    features = df[["area"]]  # 使用面积作为唯一特征
    target = df["price"]     # 预测目标：价格

    print(f"训练数据形状: 特征 {features.shape}, 目标 {target.shape}")

    # 分割数据集为训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        features, target, test_size=0.2, random_state=42  # 固定随机种子确保结果可重现
    )
    print(f"训练集: {len(X_train)} 样本, 测试集: {len(X_test)} 样本")

    # 创建并训练线性回归模型
    model = LinearRegression()
    model.fit(X_train, y_train)
    print(f"模型训练完成，系数: {model.coef_[0]:.2f}, 截距: {model.intercept_:.2f}")

    # 在测试集上进行预测
    y_pred = model.predict(X_test)

    # 计算评估指标
    metrics = {
        "r2_score": round(r2_score(y_test, y_pred), 4),  # R²分数，越接近1越好
        "mae": round(mean_absolute_error(y_test, y_pred), 2),  # 平均绝对误差
        "train_size": int(len(X_train)),  # 训练集大小
        "test_size": int(len(X_test)),    # 测试集大小
    }

    # 创建对比DataFrame，按面积排序便于可视化
    comparison_df = pd.DataFrame({
        "area": X_test["area"],
        "actual_price": y_test,
        "predicted_price": y_pred,
    }).sort_values(by="area")

    # 绘制预测值vs实际值散点图
    plt.figure(figsize=(8, 6))
    plt.scatter(comparison_df["actual_price"], comparison_df["predicted_price"], 
                c="steelblue", s=70, alpha=0.7)
    
    # 添加理想预测线（y=x）
    max_price = max(comparison_df["actual_price"].max(), comparison_df["predicted_price"].max())
    min_price = min(comparison_df["actual_price"].min(), comparison_df["predicted_price"].min())
    plt.plot([min_price, max_price], [min_price, max_price], "--", color="crimson", label="理想预测")
    
    plt.title("预测价格 vs. 实际价格")
    plt.xlabel("实际价格 (人民币)")
    plt.ylabel("预测价格 (人民币)")
    plt.legend()
    plt.tight_layout()
    plt.savefig(OUTPUT_DIR / "predicted_vs_actual.png", dpi=150)
    plt.close()
    print("已保存: predicted_vs_actual.png")

    # 保存预测样本到CSV文件
    comparison_df.to_csv(OUTPUT_DIR / "predicted_vs_actual_samples.csv", index=False)
    print("已保存: predicted_vs_actual_samples.csv")

    # 添加前5个预测样本到指标中（用于前端展示）
    metrics["latest_predictions"] = comparison_df.head(5).round(2).to_dict(orient="records")

    return metrics


def save_report(
    cleaned_df: pd.DataFrame,
    stats_df: pd.DataFrame,
    metrics: Dict[str, float],
    missing_summary: Dict[str, Dict[str, int]],
) -> None:
    """
    保存分析报告和摘要文件，供前端仪表板使用
    
    Args:
        cleaned_df: 清理后的DataFrame
        stats_df: 统计信息DataFrame
        metrics: 模型评估指标
        missing_summary: 缺失值统计摘要
    """
    # 保存统计信息到CSV
    stats_df.to_csv(OUTPUT_DIR / "summary_statistics.csv", index=True)
    print("已保存: summary_statistics.csv")

    # 计算位置分布
    location_distribution = (
        cleaned_df["location"].value_counts().sort_values(ascending=False).to_dict()
    )

    # 构建报告数据
    report_payload = {
        "record_count": int(len(cleaned_df)),  # 总记录数
        "missing_values": missing_summary,     # 缺失值统计
        "summary_statistics": {
            column: {metric: float(value) for metric, value in metrics_dict.items()}
            for column, metrics_dict in stats_df.to_dict(orient="index").items()
        },  # 统计信息
        "location_distribution": location_distribution,  # 位置分布
        "model": {
            "type": "LinearRegression",  # 模型类型
            "target": "price",           # 预测目标
            "features": ["area"],        # 使用特征
            "metrics": metrics,          # 评估指标
        },
    }

    # 处理JSON序列化中的NumPy类型
    def json_default(obj):
        if isinstance(obj, (np.floating, np.integer)):
            return obj.item()  # 将NumPy类型转换为Python原生类型
        raise TypeError(f"类型 {type(obj)} 不可序列化")

    # 保存JSON报告
    report_path = OUTPUT_DIR / "report.json"
    report_path.write_text(json.dumps(report_payload, indent=2, default=json_default), encoding="utf-8")
    print("已保存: report.json")


def main() -> None:
    """主函数：执行完整的数据分析流程"""
    # 确保输出目录存在
    OUTPUT_DIR.mkdir(exist_ok=True, parents=True)
    print("开始数据分析流程...")

    # 步骤1: 加载数据
    raw_df = load_data(DATA_PATH)
    print("=== 原始数据预览 ===")
    print(raw_df.head())

    # 步骤2: 数据清理
    cleaned_df, missing_summary = clean_data(raw_df)
    print("\n=== 缺失值统计 ===")
    for phase, summary in missing_summary.items():
        print(f"{phase}: {summary}")

    # 步骤3: 计算基本统计
    stats_df = compute_basic_statistics(cleaned_df)
    print("\n=== 统计摘要 ===")
    print(stats_df)

    # 步骤4: 生成可视化图表
    generate_visualisations(cleaned_df)
    
    # 步骤5: 训练和评估模型
    metrics = train_and_evaluate(cleaned_df)

    print("\n=== 模型性能 ===")
    for key, value in metrics.items():
        if key != "latest_predictions":  # 不打印预测样本，保持输出简洁
            print(f"{key}: {value}")

    # 步骤6: 保存报告
    save_report(cleaned_df, stats_df, metrics, missing_summary)
    print("\n所有分析结果已保存到 'outputs' 目录。")


if __name__ == "__main__":
    main()